//package tk.xboot.flink.cep;
//
//
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.cep.CEP;
//import org.apache.flink.cep.PatternSelectFunction;
//import org.apache.flink.cep.pattern.Pattern;
//import org.apache.flink.cep.pattern.conditions.SimpleCondition;
//import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.snakeyaml.events.Event;
//import org.apache.flink.streaming.api.TimeCharacteristic;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
//import org.apache.flink.streaming.api.watermark.Watermark;
//
//import java.util.List;
//import java.util.Map;
//

// https://www.cnblogs.com/fxjwind/p/8085298.html

//public class Demo {
//    public static void main(String[] args) {
//
//
//        /**
//         *
//         * 我们先产生一个输入流，这个输入Event流由Event对象和event time组成
//         * 那么要使用EventTime，除了指定TimeCharacteristic外，在Flink中还要assignTimestampsAndWatermarks，
//         * 其中分别定义了Eventtime和WaterMark，
//         */
//
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//
//        // (Event, timestamp)
//        DataStream<Event> input = env.fromElements(
//                Tuple2.of(new Event(1, "start", 1.0), 5L),
//                Tuple2.of(new Event(2, "middle", 2.0), 1L),
//                Tuple2.of(new Event(3, "end", 3.0), 3L),
//                Tuple2.of(new Event(4, "end", 4.0), 10L), //触发2，3，1
//                Tuple2.of(new Event(5, "middle", 5.0), 7L),
//                // last element for high final watermark
//                Tuple2.of(new Event(5, "middle", 5.0), 100L) //触发5，4
//        ).assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<Tuple2<Event, Long>>() {
//
//            @Override
//            public long extractTimestamp(Tuple2<Event, Long> element, long previousTimestamp) {
//                return element.f1; //定义Eventtime
//            }
//
//            @Override
//            public Watermark checkAndGetNextWatermark(Tuple2<Event, Long> lastElement, long extractedTimestamp) {
//                return new Watermark(lastElement.f1 - 5); //定义watermark
//            }
//
//        }).map(new MapFunction<Tuple2<Event, Long>, Event>() {
//            @Override
//            public Event map(Tuple2<Event, Long> value) throws Exception {
//                return value.f0;
//            }
//        });
//
//
//        /**
//         * 接着我们定义需要匹配的pattern，需求就是找出包含”start“， ”middle“， ”end“的一组事件
//         * 具体语法参考Flink文档，这里不详述
//         */
//
//
//        Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
//
//            @Override
//            public boolean filter(Event value) throws Exception {
//                return value.getName().equals("start");
//            }
//        }).followedByAny("middle").where(new SimpleCondition<Event>() {
//
//            @Override
//            public boolean filter(Event value) throws Exception {
//                return value.getName().equals("middle");
//            }
//        }).followedByAny("end").where(new SimpleCondition<Event>() {
//
//            @Override
//            public boolean filter(Event value) throws Exception {
//                return value.getName().equals("end");
//            }
//        });
//
//        /**
//         * 最终在输入流上执行CEP，
//         * 这里实现PatternSelectFunction来处理匹配到的pattern，处理逻辑是打印出匹配到的3个Event对象的id
//         */
//
//        DataStream<String> result = CEP.pattern(input, pattern).select(
//                new PatternSelectFunction<Event, String>() {
//
//                    @Override
//                    public String select(Map<String, List<Event>> pattern) {
//                        StringBuilder builder = new StringBuilder();
//                        System.out.println(pattern);
//                        builder.append(pattern.get("start").get(0).getId()).append(",")
//                                .append(pattern.get("middle").get(0).getId()).append(",")
//                                .append(pattern.get("end").get(0).getId());
//
//                        return builder.toString();
//                    }
//                }
//        );
//
//        result.print();
//
//        /**
//         * 大家想想，这里匹配到的是哪些Event？
//         * 从上面Event的顺序看应该是 1，2，3
//         * 但结果是 1，5，4，因为这里考虑的是Eventtime的顺序，这个特性在生产环境中很关键，
//         * 因为我们无法保证采集数据达到的顺序。
//         *
//         */
//
//
//    }
//}
